package main

import (
	"errors"
	"flag"
	"fmt"
	"log"
	"time"

	"github.com/nsqio/go-nsq"
)

func text(a, b int64) error {
	if a > b {
		return nil
	}
	return errors.New("信息处理错误！！！")
}

func main() {
	handler := func(message *nsq.Message) error {
		fmt.Println(string(message.Body))
		return text(1, 2)
	}
	go StartConsumer("topic_xp", "test_xp-channel", handler)
	StartProducer("topic_xp", "messageProducer_测试")
}

/*
func Nsq_CP(topic, channel, messageProducer string, handler func(message *nsq.Message) error) {
		handler := func(message *nsq.Message) error {
		fmt.Println(string(message.Body))
		return nil
	}
	go StartConsumer(topic, channel, handler)
	StartProducer(topic, messageProducer)
}
*/

var url_nsq string
var url_nsq1 string

func init() {
	//具体ip,端口根据实际情况传入或者修改默认配置
	flag.StringVar(&url_nsq, "url", "127.0.0.1:4150", "nsqd")         //tcp
	flag.StringVar(&url_nsq1, "url1", "127.0.0.1:4161", "nsqlookupd") //http
	flag.Parse()
}

// 生产者
func StartProducer(topic, messageProducer string) {
	cfg := nsq.NewConfig()
	producer, err := nsq.NewProducer(url_nsq, cfg)
	if err != nil {
		log.Fatal(err)
	}
	// 发布消息
	if err := producer.Publish(topic, []byte(messageProducer)); err != nil {
		log.Fatal("publish error: " + err.Error())
	}
	time.Sleep(1 * time.Second)
}

// 消费者
func StartConsumer(topic, channel string, handler func(message *nsq.Message) error) {
	cfg := nsq.NewConfig()
	consumer, err := nsq.NewConsumer(topic, channel, cfg)
	if err != nil {
		log.Fatal(err)
	}
	//设置消息处理函数
	consumer.AddHandler(nsq.HandlerFunc(handler))

	if err := consumer.ConnectToNSQLookupds([]string{url_nsq1}); err != nil {
		log.Fatal(err)
	}
	<-consumer.StopChan
}
